package com.codrim.cs.data.server.rpc;

import com.alibaba.dubbo.config.annotation.Service;
import com.codrim.common.utils.exception.Assert;
import com.codrim.cs.data.rpc.AffiliateDataService;
import com.codrim.cs.data.rpc.dto.param.AffiliateWholeDayDataForPeriodParamDTO;
import com.codrim.cs.data.rpc.dto.param.AffiliateWholeDayDataParamDTO;
import com.codrim.cs.data.rpc.dto.param.AffiliateWholeDayDataRoleParamDTO;
import com.codrim.cs.data.rpc.dto.param.GeneralDataPermissionParam;
import com.codrim.cs.data.rpc.dto.param.TopAffiliateConversionParamDTO;
import com.codrim.cs.data.rpc.dto.param.WholeDayDataParamDTO;
import com.codrim.cs.data.rpc.dto.result.AffiliateWholeDayDataForPeriodResultDTO;
import com.codrim.cs.data.rpc.dto.result.RealTimeDataDTO;
import com.codrim.cs.data.rpc.dto.result.TopAffiliateConversionResultDTO;
import com.codrim.cs.data.rpc.dto.result.WholeAffiliateDataDTO;
import com.codrim.cs.data.rpc.dto.result.WholeDayDataResultDTO;
import com.codrim.cs.data.rpc.enums.TimeDimension;
import com.codrim.cs.data.server.dao.RealTimeDataDao;
import com.codrim.cs.data.server.dao.entity.DailyDataRecordDO;
import com.codrim.cs.data.server.dao.entity.StatCsAffAllDO;
import com.codrim.cs.data.server.dao.entity.SummaryAffiliateData;
import com.codrim.cs.data.server.dao.entity.TopAffiliateConversionDO;
import com.codrim.cs.data.server.dao.mapper.DataRecordMapper;
import com.codrim.cs.data.server.dao.mapper.StatCsAdvAffMapper;
import com.codrim.cs.data.server.dao.mapper.StatCsAffAllMapper;
import com.codrim.cs.data.server.dao.mapper.StatCsAffOfferMapper;
import com.codrim.cs.data.server.dao.mapper.StatCsGeneralMapper;
import com.codrim.cs.data.server.dao.param.AffAllGroupByDayAndSymbolParam;
import com.codrim.cs.data.server.dao.param.AffAllParam;
import com.codrim.cs.data.server.dao.param.AffDataRoleParam;
import com.codrim.cs.data.server.dao.param.SDateAndCompanyId;
import com.codrim.cs.data.server.dao.realtime.RealTimeData;
import com.codrim.cs.data.server.subservice.DataSubService;
import com.codrim.cs.data.server.subservice.TimeZoneSubService;
import com.codrim.cs.data.server.utils.DateUtils;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

import javax.annotation.Resource;

import static com.codrim.cs.data.server.utils.TimeDimensionUtils.dateToInt;

/**
 * @author Liang.Zhuge
 * @date 28/07/2017
 */
@Service(version = AffiliateDataService.RPC_VERSION)
public class AffiliateDataServiceImpl implements AffiliateDataService {
    private static final Logger logger = LoggerFactory.getLogger(AffiliateDataServiceImpl.class);

    @Resource private StatCsAffAllMapper statCsAffAllMapper;
    @Resource private TimeZoneSubService timeZoneSubService;
    @Resource private DataSubService dataSubService;
    @Resource private DataRecordMapper dataRecordMapper;
    @Resource private RealTimeDataDao realTimeDataDao;
    @Resource private StatCsAffOfferMapper dayStatCsAffOfferMapper;
    @Resource private StatCsAdvAffMapper statCsAdvAffMapper;
    @Resource private StatCsGeneralMapper statCsGeneralMapper;

    @Override
    public List<WholeDayDataResultDTO> affiliateWholeDayData(WholeDayDataParamDTO param) {
        logger.info("Whole day data, param:{}", param);
        Assert.notNull(param, "Invalid param, can not be null");
        final Date date = param.getDate();
        final Integer companyId = param.getCompanyId();

        final Integer zone = param.getZone();
        final String sZone = timeZoneSubService.prefixTableNameForTimeZone(zone);

        Assert.notNull(date, "Invalid param, date field can not be null");
        Assert.notNull(companyId, "Invalid param, companyId field can not be null");
        Assert.notNull(sZone, "Invalid param, sZone can not be null");

        final String pattern = "yyyyMMdd";
        final String sUtcDate = DateUtils.toString(date, pattern, zone);
        logger.info("Advertiser whole day data, utc date:{}, default date:{}, company id:{}, sZone:{}",
                sUtcDate, DateUtils.toString(date, pattern), companyId, sZone);

        final List<StatCsAffAllDO> data = statCsAffAllMapper.findBySDateAndAffId(
                new SDateAndCompanyId(NumberUtils.toInt(sUtcDate), companyId), sZone);

        List<DailyDataRecordDO> dataRecords = dataRecordMapper.affiliateDailyData(
                String.valueOf(param.getCompanyId()), DateUtils.format(param.getDate(), "yyyy-MM-dd"));

        return dataSubService.wholeDayDataForAff(data, dataRecords);
    }

    @Override
    public List<WholeDayDataResultDTO> affiliateWholeDayData(AffiliateWholeDayDataParamDTO param,
            AffiliateWholeDayDataRoleParamDTO roleParam) {
        logger.info("Whole day data, param:{}, roleParam:{}", param, roleParam);
        Assert.notNull(param, "Param can not be null");

        final Integer zone = param.getZone();
        Assert.notNull(param, "Invalid param: zone can not be null");

        final AffDataRoleParam affDataRoleParam = Optional.ofNullable(roleParam).map(AffDataRoleParam::of).orElse(null);
        final AffAllParam affAllParam = AffAllParam.of(param);

        List<StatCsAffAllDO> statCsAffAllDOs = statCsGeneralMapper.affAll(affAllParam, affDataRoleParam,
                timeZoneSubService.prefixTableNameForTimeZone(zone));

        List<DailyDataRecordDO> dataRecords = dataRecordMapper.affAll(affAllParam, affDataRoleParam,
                DateUtils.toString(param.getDate(), "yyyy-MM-dd", zone));

        return dataSubService.wholeDayDataForAff(statCsAffAllDOs, dataRecords);
    }

    @Override
    public List<RealTimeDataDTO> affiliateWholeDayRealData(WholeDayDataParamDTO param) {
        Assert.notNull(param, "Invalid param, can not be null");

        final Integer companyId = param.getCompanyId();
        Date date = param.getDate();

        Assert.notNull(companyId, "Invalid param, companyId field can not be null");
        Assert.notNull(date, "Invalid param, date can be null");

        Instant instant = date.toInstant();
        Instant begin = instant.truncatedTo(ChronoUnit.DAYS);
        Instant end = begin.plus(23, ChronoUnit.DAYS)
                .plus(59, ChronoUnit.HOURS)
                .plus(59, ChronoUnit.SECONDS);

        return realTimeDataDao.affiliateRealTimeData(companyId, begin, end)
                .stream().map(RealTimeData::toDTO).collect(Collectors.toList());
    }

    @Override
    public List<WholeAffiliateDataDTO> affiliateWholeDataForOffer(Integer offerId, Integer zoneId) {
        List<SummaryAffiliateData> summaryAffiliateDatas = dayStatCsAffOfferMapper.summaryAffiliateDataByOffer(
                offerId, timeZoneSubService.prefixTableNameForTimeZone(zoneId));
        return Optional.ofNullable(summaryAffiliateDatas)
                .map(datas -> datas.stream().map(SummaryAffiliateData::toWholeAffilidateData).collect(Collectors.toList()))
                .orElse(null);
    }

    @Override
    public List<TopAffiliateConversionResultDTO> topAffiliateConversions(TopAffiliateConversionParamDTO param) {
        Assert.notNull(param.getAdvertiserId(), "Invalid advertiser id");
        Assert.notNull(param.getBegin(), "Invalid begin date");
        Assert.notNull(param.getEnd(), "Invalid end date");

        List<TopAffiliateConversionDO> dos = statCsAdvAffMapper.topAffiliateConversions(
                param, DateUtils.dateToInt(param.getBegin()),
                DateUtils.dateToInt(param.getEnd()),
                timeZoneSubService.prefixTableNameForTimeZone(param.getTimeZone()));

        return Optional.ofNullable(dos)
                .map(List::stream)
                .map(stream -> stream.filter(Objects::nonNull))
                .map(stream -> stream.map(TopAffiliateConversionDO::toDto))
                .map(stream -> stream.collect(Collectors.toList()))
                .orElse(null);
    }

    @Override
    public AffiliateWholeDayDataForPeriodResultDTO wholeDayDataForPeriod(
            AffiliateWholeDayDataForPeriodParamDTO param, GeneralDataPermissionParam roleParam) {
        final Date begin = param.getBegin();
        final Date end = param.getEnd();
        final Integer affiliateId = param.getCompanyId();
        final Integer zone = param.getZone();
        logger.info("Params, affiliateId:{}, sZone:{}, begin:{}, end:{}", affiliateId, zone, begin, end);
        final AffiliateWholeDayDataForPeriodResultDTO result = new AffiliateWholeDayDataForPeriodResultDTO();
        boolean isSearchOneDay = begin.getDate() == end.getDate() && begin.getMonth() == end.getMonth();
        TimeDimension timeDimension = isSearchOneDay ? TimeDimension.Hour : TimeDimension.Day;

        final int[] timePeriod = timePeriod(begin, end, zone, timeDimension);
        final int iBegin = timePeriod[0];
        final int iEnd = timePeriod[1];
        logger.info("Time period, affiliateId:{}, timePeriod:{}", affiliateId, timePeriod);

        int iNow = dateToInt(org.apache.commons.lang3.time.DateUtils.addHours(new Date(), zone - 8), timeDimension);

        final AffAllGroupByDayAndSymbolParam paramDo = new AffAllGroupByDayAndSymbolParam();
        paramDo.setCompanyId(affiliateId);
        paramDo.setBegin(iBegin);
        paramDo.setEnd(iEnd);
        paramDo.setAdvertiserManagerId(param.getAdvertiserManagerId());
        paramDo.setDepartmentIds(param.getDepartmentIds());

        AffDataRoleParam roleParamDo = AffDataRoleParam.of(roleParam);
        List<StatCsAffAllDO> datas = statCsGeneralMapper.affAllGroupByDayAndSymbol(
                paramDo, roleParamDo, timeZoneSubService.prefixTableNameForTimeZone(zone),
                timeDimension.key);
        logger.info("iBegin:{}, iNow:{}, iEnd:{}, data size:{}", iBegin, iNow, iEnd, Optional.ofNullable(datas).map(List::size).orElse(0));

        if (CollectionUtils.isNotEmpty(datas)) {
            if (iNow >= iBegin && iNow <= iEnd) {
                logger.info("Include now, try to get real time data");
                List<RealTimeData> realTimeDatas = realTimeDataDao.affiliateRealTimeData(affiliateId, begin.toInstant(), end.toInstant());
                boolean nowData = datas.stream().anyMatch(item -> item.getSdate() == iNow);
                if (CollectionUtils.isNotEmpty(realTimeDatas) && nowData) {
                    logger.info("Real time data exists");
                    List<RealTimeDataDTO> realTimeDto = realTimeDatas.stream()
                            .map(RealTimeData::toDTO)
                            .collect(Collectors.toList());
                    result.setRealTimeDataDTOList(realTimeDto);
                }
            }

            List<WholeDayDataResultDTO> wholeDayDataResultDTOS = datas.stream()
                    .map(StatCsAffAllDO::toWholeDayDataResultDTO)
                    .collect(Collectors.toList());
            result.union(wholeDayDataResultDTOS);
        }

        if (timeDimension == TimeDimension.Day) {
            // 查出录入数据。因为录入数据只归属到天，仅天粒度下处理录入数据
            final List<DailyDataRecordDO> mergedDataRecord = dataRecordMapper.affAllGroupBySdateAndSymbol(paramDo,
                    roleParamDo);

            final List<WholeDayDataResultDTO> dataRecordWholeDayDataResultDTO = Optional.ofNullable(mergedDataRecord)
                    .map(List::stream)
                    .map(s -> s.map(DailyDataRecordDO::toWholeDayDataResultDTO))
                    .map(s -> s.collect(Collectors.toList()))
                    .orElse(null);
            result.union(dataRecordWholeDayDataResultDTO);
        }

        return result;
    }

    private int[] timePeriod(Date begin, Date end, Integer zone, TimeDimension timeDimension) {
        int[] timePeriod = new int[2];

        /*
         * 处理比较难理解,可参考上文示例说明(和前端参数格式绑定在了一起)
         */
        Date targetBeginDate = org.apache.commons.lang3.time.DateUtils.addHours(begin, zone - 8);
        /* 统计区间从目标时区 0时 点开始 */
        targetBeginDate.setHours(0);
        timePeriod[0] = dateToInt(targetBeginDate, timeDimension);
        if (timeDimension != TimeDimension.Hour) {
            targetBeginDate = org.apache.commons.lang3.time.DateUtils.addHours(end, zone - 8);
        }
        targetBeginDate.setHours(23);
        timePeriod[1] = dateToInt(targetBeginDate, timeDimension);
        return timePeriod;
    }
}
